package pku;

import java.io.File;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生产者
 */
public class Producer {

    private DefaultMessage message = new DefaultMessage(null);

    private String thread;

    private static AtomicInteger count = new AtomicInteger(-1);

    public Producer() {
        this.thread = count.incrementAndGet() + "";
    }

    //生成一个指定topic的message返回
    public ByteMessage createBytesMessageToTopic(String topic, byte[] body) {
        message.clear();
        message.putHeaders(MessageHeader.TOPIC, topic);
        message.setBody(body);
        return message;
    }

    private Map<String, MessageWriter> writers = new HashMap<>();

    //将message发送出去
    public void send(ByteMessage defaultMessage) {

        try {
            String topic = defaultMessage.headers().getString(MessageHeader.TOPIC);
            MessageWriter writer = writers.get(topic);
            if (writer == null) {
                String path = Constants.DIR + File.separator + topic;
                File dir = new File(path);
                if (!dir.exists()) {
                    dir.mkdir();
                }

                writer = new MessageWriter(path + File.separator + "t" + thread);
                writers.put(topic, writer);

            }
            writer.write((DefaultMessage) defaultMessage);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //处理将缓存区的剩余部分
    public void flush() throws Exception {
        try {
            for (Map.Entry<String, MessageWriter> e : writers.entrySet()) {
                e.getValue().close();
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
